package net.bwie.realtime.jtp.dws.log.fuction;

import net.bwie.realtime.jtp.utils.DateTimeUtil;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

public class PageViewReportWindowFunction
    implements WindowFunction<PageViewBean,String,String, TimeWindow> {
    @Override
    public void apply(
            String key,
            TimeWindow window,
            Iterable<PageViewBean> input,
            Collector<String> out) throws Exception {
        /*
        key: 表示分组字段, window: 表示时间窗口,可以获取窗口开始和结束时间, input: 表示窗口中所有数据
         */
        // 1.窗口时间
        String windowStart = DateTimeUtil.convertLongToString(window.getStart(), "yyyy-MM-dd HH:mm:ss");
        String windowEnd = DateTimeUtil.convertLongToString(window.getEnd(), "yyyy-MM-dd HH:mm:ss");
        // 2.获取窗口中计算结果
        PageViewBean bean = input.iterator().next();
        Long pvCount = bean.getPvCount();
        Long pvDuration = bean.getPvDuration();
        Long uvCount = bean.getUvCount();
        Long sessionCount = bean.getSessionCount();
        // 3.输出
        String output = windowStart + "," + windowEnd + "," + key + "," + pvCount
                + "," + pvDuration + "," + uvCount + "," + sessionCount + "," + System.currentTimeMillis();
        out.collect(output);
    }
}
